package com.example.demo.service;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.example.demo.dao.PositionDao;
import com.example.demo.entity.Position;
import com.example.demo.util.DBHelper;
import com.example.demo.util.EsUtil;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.FuzzyQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.sql.*;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
@Slf4j
public class EsServiceImpl implements EsService {

    @Qualifier("esClient")
    @Autowired
    RestHighLevelClient client;
    @Autowired
    PositionDao positionDao;

    private static final String POSITION_INDEX = "position";


    @Override
    public List<SearchHit> getPositionByName(String name) throws IOException {

        List<SearchHit> collect = searchHitList(POSITION_INDEX, "position_name", name);
        if (CollectionUtils.isEmpty(collect)) {
            //发起第二次查询，凑够5条数据，position_advantage 查找 美女多、员工福利好 的企业职位信息进行补充够5条）
            List<SearchHit> positionAdvantage = searchHitList(POSITION_INDEX, "position_advantage", "美女多、员工福利好");
            if (!CollectionUtils.isEmpty(positionAdvantage)) {
                return positionAdvantage.stream().limit(5).collect(Collectors.toList());
            }
        }
        List<SearchHit> searchHitList=new ArrayList<>(collect);
        //查询结果的数量
        int size = collect.size();
        if (size >= 5) {
            return collect;
        }
        int num = 5 - size;
        List<SearchHit> positionAdvantage = searchHitList(POSITION_INDEX, "position_advantage", "美女多、员工福利好");
        if (!CollectionUtils.isEmpty(positionAdvantage)) {
            List<SearchHit> advantageList = positionAdvantage.stream().limit(num).collect(Collectors.toList());
            searchHitList.addAll(advantageList);
        }
        return searchHitList;
    }


    private List<SearchHit> searchHitList(String index, String field, Object value) throws IOException {
        SearchRequest searchRequest = new SearchRequest(index);
        //构建搜索条件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        MatchQueryBuilder matchQueryBuilder = QueryBuilders.matchQuery(field, value);
        searchSourceBuilder.query(matchQueryBuilder);
        searchRequest.source(searchSourceBuilder);
        SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT);
        return Arrays.stream(search.getHits().getHits()).collect(Collectors.toList());
    }


    @Override
    public void importPositionToES(String tableName) {
        BulkProcessor bulkProcessor = EsUtil.getBulkProcessor(client);
        Connection connection = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            connection = DBHelper.getConn();
            log.info("start handle data :" + tableName);
            String sql = "select * from " + tableName;
            ps = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            // 根据自己需要设置 fetchSize
            ps.setFetchSize(20);
            rs = ps.executeQuery();
            ResultSetMetaData colData = rs.getMetaData();
            List<HashMap<String, String>> dataList = new ArrayList<>();
            HashMap<String, String> map = null;
            int count = 0;
            // c 就是列的名字   v 就是列对应的值
            String c = null;
            String v = null;
            while (rs.next()) {
                count++;
                map = new HashMap<String, String>(128);
                for (int i = 1; i < colData.getColumnCount(); i++) {
                    c = colData.getColumnName(i);
                    v = rs.getString(c);
                    map.put(c, v);
                }
                dataList.add(map);
                // 每1万条 写一次   不足的批次的数据 最后一次提交处理
                if (count % 10000 == 0) {
                    log.info("mysql handle data  number:" + count);
                    // 将数据添加到 bulkProcessor
                    for (HashMap<String, String> hashMap2 : dataList) {
                        bulkProcessor.add(new IndexRequest(POSITION_INDEX).source(hashMap2));
                    }
                    // 每提交一次 清空 map 和  dataList
                    map.clear();
                    dataList.clear();
                }
            }
            // 处理 未提交的数据
            for (HashMap<String, String> hashMap2 : dataList) {
                bulkProcessor.add(new IndexRequest(POSITION_INDEX).source(hashMap2));
            }
            bulkProcessor.flush();

        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            try {
                rs.close();
                ps.close();
                connection.close();
                boolean terinaFlag = bulkProcessor.awaitClose(150L, TimeUnit.SECONDS);
                log.info(String.valueOf(terinaFlag));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }
}
